package com.intct.func;

import com.alibaba.fastjson.JSONObject;
import com.intct.hbase.bean.DimConfigBean;
import com.intct.util.HBaseUtil;
import com.intct.util.JDBCUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author gufg
 * @since 2025-07-03 11:41
 */
public class DimBroadcastFunc extends BroadcastProcessFunction<String, String, JSONObject> {

    /**
     * 声明MapStateDescriptor
     */
    private MapStateDescriptor mapStateDescriptor;

    private Map<String, DimConfigBean> configMap = new HashMap<>();

    /**
     * 类构造器
     * @param mapStateDescriptor
     */
    public DimBroadcastFunc(MapStateDescriptor mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        Connection connection = JDBCUtil.getMySqlConnection();
        List<DimConfigBean> query = JDBCUtil.query(connection, "select * from travel_config.dim_table_config", DimConfigBean.class);
        for (DimConfigBean dimConfigBean : query) {
            configMap.put(dimConfigBean.getSourceTable(), dimConfigBean);
            // 创建HBase表
            HBaseUtil.createTable(dimConfigBean.getSinkTable(), dimConfigBean.getSinkFamily());
        }
        JDBCUtil.closeMySqlConnection(connection);
    }

    /**
     * 主流数据处理
     * @param jsonStr The stream element.
     * @param ctx A {@link ReadOnlyContext} that allows querying the timestamp of the element,
     *     querying the current processing/event time and updating the broadcast state. The context
     *     is only valid during the invocation of this method, do not store it.
     * @param out The collector to emit resulting elements to
     * @throws Exception
     */
    @Override
    public void processElement(String jsonStr, BroadcastProcessFunction<String, String, JSONObject>.ReadOnlyContext ctx,
                               Collector<JSONObject> out) throws Exception {
//        System.out.println("processElement开始。。。。。。");
        // 获取广播状态
        ReadOnlyBroadcastState<String, DimConfigBean> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        // 主流--kafka
        // {"op":"u","before":{"code":110101,"name":"北京市东城区44"},"after":{"code":110101,"name":"北京市东城区55"},
        // "ts_ms":1751507332360,"table":"area_info"}
        JSONObject jsonObj = JSONObject.parseObject(jsonStr);
        // 获取表名称
        String tableName = jsonObj.getString("table");

        // 使用主流中表与广播状态获取对应
        DimConfigBean dimConfig = broadcastState.get(tableName);

        // 如果两个流中同时存在
        if (dimConfig != null || configMap != null ) {
            // 主流JSON字符串
            JSONObject afterObj = jsonObj.getJSONObject("after");

            // 广播状态中获取字段列表
            String sinkColumns = dimConfig != null ? dimConfig.getSinkColumns() : configMap.get(tableName).getSinkColumns();

            // 字段过滤
            filterColumn(afterObj, sinkColumns);

            // 操作类型
            afterObj.put("op", jsonObj.getString("op"));
            // sink 表名称
            afterObj.put("sinkTable", configMap.get(tableName).getSinkTable());
            // 列族名称
            afterObj.put("sinkFamily", configMap.get(tableName).getSinkFamily());
            // 主键字段名称
            afterObj.put("sinkRowKey", configMap.get(tableName).getSinkRowKey());

            // 写出数据
            out.collect(afterObj);
//            System.out.println("processElement写出。。。。。。");
        }
//        System.out.println("processElement结束。。。。。。");
    }

    /**
     * 广播状态处理
     * @param value The stream element.
     * @param ctx A {@link Context} that allows querying the timestamp of the element, querying the
     *     current processing/event time and updating the broadcast state. The context is only valid
     *     during the invocation of this method, do not store it.
     * @param out The collector to emit resulting elements to
     * @throws Exception
     */
    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<String, String, JSONObject>.Context ctx,
                                        Collector<JSONObject> out) throws Exception {
//        System.out.println("processBroadcastElement开始。。。。。。");
        // 获取广播状态
        BroadcastState broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        // 转为JSONObject
        JSONObject jsonObj = JSONObject.parseObject(value);
        // 获取after中数据
        JSONObject afterObj = jsonObj.getJSONObject("after");

        // 业务操作类型
        String op = jsonObj.getString("op");
        // 组成JavaBean
        DimConfigBean dimConfig = DimConfigBean.builder()
                .sourceTable(afterObj.getString("source_table"))
                .sinkTable(afterObj.getString("sink_table"))
                .sinkFamily(afterObj.getString("sink_family"))
                .sinkColumns(afterObj.getString("sink_columns"))
                .sinkRowKey(afterObj.getString("sink_row_key"))
                .build();
        // 如果业务数据不是删除
        if (!op.equals("d")) {
            // 将数据放到状态中
            broadcastState.put(dimConfig.getSourceTable(), dimConfig);
            configMap.put(dimConfig.getSourceTable(), dimConfig);
        } else {
            broadcastState.remove(dimConfig.getSourceTable());
            configMap.remove(dimConfig.getSourceTable());
        }
//        System.out.println("processBroadcastElement结束。。。。。。");
    }

    /**
     * 过滤掉主流不需要的字段
     * @param afterObj
     * @param sinkColumns
     */
    private void filterColumn(JSONObject afterObj, String sinkColumns) {
        // 将广播状态流中字段拆分
        String[] broadcatColumns = sinkColumns.split(",");
        // 转换为列表
        List<String> columnList = Arrays.asList(broadcatColumns);
        // 如果主流中字段不存在于广播状态流中，则删除字段
        afterObj.entrySet().removeIf(e -> !columnList.contains(e.getKey()));
    }
}
